package com.hzh.flink.HomeWork

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.time.Duration

object Demo2Work2 {
  def main(args: Array[String]): Unit = {

    //创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    /**
     * 构建kafka source,读取cars数据
     *  1、从kafka中读取卡口过车数据
     */
    val source: KafkaSource[String] = KafkaSource
      .builder[String]
      .setBootstrapServers("master:9092,node1:9092,node2:9092") //kafka集群broker列表
      .setTopics("cars") //指定topic
      .setGroupId("carsGroup") //指定消费者组，一条数据在一个组内只被消费一次
      .setStartingOffsets(OffsetsInitializer.earliest()) //读取数据的位置，earliest：读取所有的数据，latest：读取最新的数据
      .setValueOnlyDeserializer(new SimpleStringSchema()) //反序列的类
      .build

    //使用kafka source
    val kafkaDS: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")


    /**
     * 1、解析json格式的数据
     */

    val cardAndTimeDS: DataStream[(Long, Long)] = kafkaDS.map(line => {
      //使用fastjson工具解析json格式的数据
      val cardJson: JSONObject = JSON.parseObject(line)
      //取出编号
      val card: Long = cardJson.getLong("card")
      //事件时间，事件时间要求时毫秒级别
      val time: Long = cardJson.getLong("time") * 1000
      //返回元组
      (card, time)
    })

    /**
     * 设置时间字段和水位线
     *
     */

    val assDS: DataStream[(Long, Long)] = cardAndTimeDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        //设置水位线的生成策略，前移5秒
        .forBoundedOutOfOrderness(Duration.ofSeconds(5))
        //设置时间字段
        .withTimestampAssigner(new SerializableTimestampAssigner[(Long, Long)] {
          override def extractTimestamp(element: (Long, Long), recordTimestamp: Long): Long = {
            //时间字段
            element._2
          }
        })
    )

    /**
     * 统计车流量
     *
     */
    val cardTime2DS: DataStream[(Long, Int)] = assDS.map(kv => (kv._1, 1))

    //按照卡口分组
    val cardTimeKBDS: KeyedStream[(Long, Int), Long] = cardTime2DS.keyBy(_._1)

    //开窗口
    val cardTimeWindow: WindowedStream[(Long, Int), Long, TimeWindow] = cardTimeKBDS
      .window(SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(5)))

    val countDS: DataStream[(Long, Int)] = cardTimeWindow.sum(1)

    countDS.print()

    /**
     *  将统计结果保存到mysql中
     */

    countDS.addSink(new RichSinkFunction[(Long, Int)] {


      //插入数据
      override def invoke(value: (Long, Int), context: SinkFunction.Context): Unit = {
        println("数据写入mysql")
        stat.setLong(1, value._1)
        stat.setInt(2, value._2)

        stat.execute()
      }


      var con: Connection = _
      var stat: PreparedStatement = _

      //创建链接
      override def open(parameters: Configuration): Unit = {
        //1、加载驱动
        Class.forName("com.mysql.jdbc.Driver")
        //创建链接
        con = DriverManager.getConnection("jdbc:mysql://master:3306/test", "root", "123456")
        //编写插入数据的sql
        //replace :如果不存在插入，如果存在就替换，需要在表中设置主键
        stat = con.prepareStatement("replace into card_count(card,count) values(?,?)")
      }


      //关闭链接
      override def close(): Unit = {
        stat.close()
        con.close()
      }

    })

    env.execute()

  }
}

